package com.androwit.engine.cassandra;

import java.util.HashMap;
import java.util.Map;

import me.prettyprint.cassandra.model.BasicColumnFamilyDefinition;
import me.prettyprint.cassandra.service.ThriftCfDef;
import me.prettyprint.hector.api.ddl.ColumnFamilyDefinition;
import me.prettyprint.hector.api.ddl.ColumnType;
import me.prettyprint.hector.api.ddl.ComparatorType;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.androwit.engine.mapping.AbstractMappingFieldsDefiner;
import com.androwit.engine.mapping.MappingField;

public abstract class HectorMappingFieldsDefiner<J, K, V> extends
		AbstractMappingFieldsDefiner<J, K, V> {
	private final static Logger logger = LoggerFactory
			.getLogger(HectorMappingFieldsDefiner.class);

	/**
	 * Cree une nouvelle instance de HectorMappingFieldsDefiner.
	 * 
	 * @param superInstances
	 */
	public HectorMappingFieldsDefiner(
			AbstractMappingFieldsDefiner<J, K, V>... superInstances) {
		super(superInstances);
	}

	/**
	 * returns the column family name
	 * 
	 * @return column family name
	 */
	protected abstract String getColumnFamily();

	/**
	 * creates the cassandra key for input
	 * 
	 * @param input
	 *            object to persist
	 * @return key
	 */
	public abstract K getKey(Object input);

	@SuppressWarnings("rawtypes")
	public static ColumnFamilyDefinition getColumnDefinition(
			String keyspaceName, MappingField[] fields, String columnFamily) {
		fields = removeDuplicates(fields);

		BasicColumnFamilyDefinition cfDefinitionBasic = new BasicColumnFamilyDefinition();
		cfDefinitionBasic.setKeyspaceName(keyspaceName);
		cfDefinitionBasic.setName(columnFamily);
		if (fields != null) {
			for (MappingField mappingField : fields) {
				HectorMappingField hectorMappingField = (HectorMappingField) mappingField;
				cfDefinitionBasic.addColumnDefinition(hectorMappingField
						.getColumnDefinition(columnFamily));
				logger.debug(columnFamily
						+ ": field "
						+ mappingField
						+ " is mapped"
						+ (hectorMappingField.isSecondaryIndex() ? " and INDEXED."
								: "."));
				System.out
						.println(columnFamily
								+ ": field "
								+ mappingField
								+ " is mapped"
								+ (hectorMappingField.isSecondaryIndex() ? " and INDEXED."
										: "."));
			}
		}
		cfDefinitionBasic.setColumnType(ColumnType.STANDARD);
		cfDefinitionBasic.setComparatorType(ComparatorType.UTF8TYPE);
		ColumnFamilyDefinition cfDefinition = new ThriftCfDef(cfDefinitionBasic);
		return cfDefinition;
	}

	/**
	 * removeDuplicates
	 * 
	 * @param fields
	 * @return
	 */
	@SuppressWarnings("rawtypes")
	private static MappingField[] removeDuplicates(MappingField[] fields) {
		Map<Object, MappingField> map = new HashMap<Object, MappingField>();
		for (MappingField mappingField : fields) {
			Object key = mappingField.getKey();
			if (map.containsKey(key)) {
				logger.debug("Key " + key + " was duplicate");
				continue;
			}
			map.put(key, mappingField);
		}
		return map.values().toArray(new MappingField[0]);
	}

}
